package com.huawei.codearts.tools.tool;

import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.PutResponse;
import io.etcd.jetcd.lease.LeaseGrantResponse;
import io.etcd.jetcd.lock.LockResponse;
import io.grpc.stub.StreamObserver;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 此类是测试demo 不够完善
 * 参考https://blog.csdn.net/ltx1472580369/article/details/113808965
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class EtcdLock {
    private final static Logger LOGGER = LoggerFactory.getLogger(EtcdLock.class);
//    private static final String ENDPOINTS = "http://localhost:2379";
    private static final ByteSequence NAMESPACE = ByteSequence.from("my_locks", StandardCharsets.UTF_8);
    private String lockPath;
    private long leaseId;
    // create client
//    Client client = Client.builder().endpoints(ENDPOINTS.split(",")).build();
    private final Client etcdClient;

    public void lock(String[] args) throws ExecutionException, InterruptedException {
        // 初始化etcd客户端
//        Client client = Client.builder().endpoints(ENDPOINTS.split(",")).build();
//        KV kv = client.getKVClient();
        Thread currentThread = Thread.currentThread();
        Lock lockClient = etcdClient.getLockClient();
        Lease leaseClient = etcdClient.getLeaseClient();

        // 尝试获取锁
        String lockKey = "/my_lock";
        long lockAcquireTime = 10; // 锁的最大获取时间（秒）
        try {
            // 创建一个租约
            LeaseGrantResponse ttl = leaseClient.grant(TimeUnit.NANOSECONDS.toSeconds(lockAcquireTime)).get();
            leaseId = ttl.getID();

            // 自动续约-防止ttl超时线程未执行完
            leaseClient.keepAlive(leaseId, new StreamObserver() {
                @Override
                public void onNext(Object value) {
                    //执行续约之后的回调
                    log.info("执行续约...ID: {}, ttl: {}", leaseId, ttl.getTTL());
                }

                @Override
                public void onError(Throwable t) {
                    //异常，就会停止续约，比如调用revoke取消续约，租约不存在
                    LOGGER.info("停止续约...{} 原因: {}", leaseId, t.getMessage());
                }

                @Override
                public void onCompleted() {
                    LOGGER.info("onCompleted...");
                }
            });

            // 使用etcd的分布式锁API
//            LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get(lockAcquireTime, TimeUnit.SECONDS);
            LockResponse lockResponse = lockClient.lock(ByteSequence.from(lockKey.getBytes()), leaseId).get();
            if (lockResponse != null) {
                long start = System.currentTimeMillis();
                lockPath = lockResponse.getKey().toString(StandardCharsets.UTF_8);
                LOGGER.info("线程：{} 加锁成功，锁路径：{}", currentThread.getName(), lockPath);
                System.out.println("Lock acquired");

                // 模拟线程执行时长
                TimeUnit.SECONDS.sleep(3);
                LOGGER.info("执行时长... {} ms", System.currentTimeMillis() - start);

            } else {
                System.out.println("Failed to acquire lock");
            }
        }catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            // 释放锁
            if (lockPath != null) {
                lockClient.unlock(ByteSequence.from(lockPath.getBytes())).get();
                System.out.println("Lock released");
            }
            // 删除租约
            if (leaseId != 0L) {
                leaseClient.revoke(leaseId);
            }
            // 关闭客户端
//            etcdClient.close();
        }
    }
}
